package com.heima.kafka.config;

import lombok.Getter;
import lombok.Setter;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.KafkaStreamsConfiguration;

import java.util.HashMap;
import java.util.Map;

/**
 * 通过重新注册KafkaStreamsConfiguration对象，设置自定配置参数
 */

@Setter
@Getter
@Configuration
@EnableKafkaStreams
@ConfigurationProperties(prefix="kafka")
public class KafkaStreamConfig {
    private static final int MAX_MESSAGE_SIZE = 16* 1024 * 1024;
    private String hosts;
    private String group;
    /**
     * 创建并配置一个KafkaStreams的默认配置
     *
     * 该方法定义了一个Spring的@Bean方法，用于创建和配置KafkaStreams的默认配置实例
     * 它设置了KafkaStreams启动时所需的基本配置，包括服务器地址、应用ID、客户端ID、重试次数以及序列化类
     *
     * @return KafkaStreamsConfiguration 返回配置好的KafkaStreams配置实例
     */
    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {
        // 初始化配置属性的Map
        Map<String, Object> props = new HashMap<>();
        // 设置Kafka服务器的地址列表
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, hosts);
        // 设置KafkaStreams应用的ID，用于标识应用
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, this.getGroup()+"_stream_aid");
        // 设置KafkaStreams客户端的ID，用于监控和跟踪
        props.put(StreamsConfig.CLIENT_ID_CONFIG, this.getGroup()+"_stream_cid");
        // 设置重试配置，当生产者收到临时错误时自动重试的次数
        props.put(StreamsConfig.RETRIES_CONFIG, 10);
        // 设置默认的键序列化类
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        // 设置默认的值序列化类
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        // 返回配置好的KafkaStreams配置实例
        return new KafkaStreamsConfiguration(props);
    }

}